package com.tjbklx1.rocketMQTest.demo01;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import com.tjbklx1.rocketMQTest.constant.Host;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

public class PullConsumer {
	private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<>();

	public static void main(String[] args) {
		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(Host.PULL_CONSUMER_GROUP);
		consumer.setInstanceName(UUID.randomUUID().toString());
		consumer.setMessageModel(MessageModel.BROADCASTING);
		consumer.setNamesrvAddr(Host.HOST);
		try {
			consumer.start();
		} catch (MQClientException e) {
			e.printStackTrace();
		}
		try {
			Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(Host.TOPIC);
			for (MessageQueue mq : mqs) {
				System.out.println("Consume from the queue: " + mq + "%n");
				SINGLE_MQ: while (true) {
					try {
						PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
						System.out.printf("%s%n", pullResult);
						putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
						switch (pullResult.getPullStatus()) {
						case FOUND:
							break;
						case NO_MATCHED_MSG:
							break;
						case NO_NEW_MSG:
							break SINGLE_MQ;
						case OFFSET_ILLEGAL:
							break;
						default:
							break;
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		} catch (MQClientException e) {
			e.printStackTrace();
		}
	}

	private static long getMessageQueueOffset(MessageQueue mq) {
		Long offset = OFFSE_TABLE.get(mq);
		if (offset != null)
			return offset;
		return 0;
	}

	private static void putMessageQueueOffset(MessageQueue mq, long offset) {
		OFFSE_TABLE.put(mq, offset);
	}
}